package interview.blockqueue;


import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/***
 *   阻塞队列版的生产者和消费者
 *
 * */

class ShareResource{

    private  volatile boolean FLAG = true;
    private AtomicInteger atomicInteger = new AtomicInteger();
    BlockingQueue<Integer> blockingQueue = null;

    public ShareResource(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public  void product(){
        System.out.println("生产线程启动");
        try {
            while (FLAG) {
                int data=atomicInteger.incrementAndGet();
              boolean ret = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
              if(ret){
                  System.out.println(Thread.currentThread().getName()+"加入队列"+data+"成功");
              }else{
                  System.out.println(Thread.currentThread().getName()+"加入队列"+data+"失败");
                  //FLAG = false;
              }
                TimeUnit.SECONDS.sleep(3);
              }
            System.out.println("主线程叫停，生产停止");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
    public void consumer(){
        System.out.println("消费线程启动");
        try {
            while (FLAG) {
                Integer ret =  blockingQueue.poll(2L, TimeUnit.SECONDS);
                if (ret==null){
                    System.out.println(Thread.currentThread().getName()+"超过两秒未等到结果，消费者退出");
                    FLAG = false;
                }else{
                    System.out.println("     "+Thread.currentThread().getName()+"消费队列"+ret+"成功");
                }
                TimeUnit.SECONDS.sleep(1);
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }finally {
            System.out.println(Thread.currentThread().getName()+"停止");
        }
    }
    public  void stop(){
        this.FLAG = false;
    }
}

public class BlockQueueProductAndConsumer {

    public static void main(String[] args) throws InterruptedException {

    ShareResource shareResource = new ShareResource(new ArrayBlockingQueue(3));
    new Thread(()->{
        shareResource.product();
    },"生产线程").start();

    new Thread(()->{
        shareResource.consumer();
    },"消费线程").start();

        TimeUnit.SECONDS.sleep(10);
        System.out.println("5秒时间到，主线程停止");
        shareResource.stop();
    }
}